package com.flute.haflute.agent;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.jdom.JDOMException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.agent.heartbeat.HeartBeatGuard;
import com.flute.haflute.agent.heartbeat.HeartBeatWatcher;
import com.flute.haflute.agent.manage.GeneralClusterManager;
import com.flute.haflute.agent.manage.JvmManager;
import com.flute.haflute.agent.topo.ClusterTopoDBService;
import com.flute.haflute.agent.topo.Node;
import com.flute.haflute.common.ClusterConfLoader;
import com.flute.haflute.common.ClusterConstants;
import com.flute.haflute.common.ClusterConstants.ROLE;
import com.flute.haflute.common.ClusterContext;
import com.flute.haflute.jobbox.base.RunProcessInterface;
import com.flute.haflute.jobbox.common.RunProcessLauncher;
import com.flute.haflute.monitor.ClusterStatusMonitor;
import com.flute.haflute.monitor.JobMonitor;
import com.flute.haflute.monitor.LocalStatusMonitor;
import com.flute.haflute.monitor.UniversalMonitorMsgSender;
import com.flute.haflute.net.RMILocalService;
import com.flute.haflute.net.RMINetService;
import com.flute.haflute.net.messaging.MessagingService;
import com.flute.haflute.tools.ClusterUtils;
import com.flute.haflute.web.JettyServer;

/**
 * 平台代理启动主线程，初始化系统环境、读取配置参数、配置相关模块、启动后台线程。<br>
 * 
 * Haflute分布式集群平台每一个机器节点上的守护进程称为Agent（代理），Agent负责所有节点间的通信联系，由此组成一颗集群树。<br>
 * 
 * 集群节点在集群中的角色包括:<br>
 * 
 * DM：指定主节点，集群的根控制节点<br>
 * BDM：备用主节点，集群的备份根节点<br>
 * SM：选举主节点，根据集群扩展和工作调度时，管理部分SLAVE节点，并向DM/BDM汇报的节点 <br>
 * SLAVE：从节点，为普通节点，向DM/BDM或者SM汇报信息，执行任务，为集群任务执行成员<br>
 * 
 * Agent的功能：进行集群沟通和角色、状态维护，保障集群状态的正确性和一致性；跟踪所有集群和工作信息；容错和故障恢复。
 * 
 */
public class Haflute {
	private static Logger logger = LoggerFactory.getLogger(Haflute.class);

	// 日志存储的配置文件，log4j
	private static String logConfigFile = "conf/agent.lcf";

	public static void main(String[] args) {

		System.err.println("starting Agent...");

		try {

			// 初始化上下文环境: 1. 日志配置 2. 集群参数导入
			initContext(args);

			// 根据配置和命令行参数解析当前Agent在集群中角色
			registerRoleInCluster(args);

			// 启动网络IO相关的线程
			startNetService(args);

			// 启动所有后台线程
			startBackends(args);

			startMonitorModule(args);

			setDefaultExceptionHandler();

			// 向系统注册Hook，系统关闭时清理环境
			Runtime.getRuntime().addShutdownHook(new ShutdownHook());

		} catch (Exception e) {
			e.printStackTrace();
			logger.error("", e);
			ClusterUtils.exitGracefully();
		}

		System.err.println("Cluster Agent successfully started!");
	}

	/**
	 * 抓住没有捕获的异常，防止导致JVM崩溃而无法确定原因
	 */
	private static void setDefaultExceptionHandler() {

		Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
			public void uncaughtException(Thread t, Throwable e) {
				logger.error("Fatal exception in thread " + t.getName(), e);
				if (e instanceof OutOfMemoryError) {
					ClusterUtils.exitGracefully();
				}
			}
		});

	}

	/**
	 * 启动所有后台线程
	 * 
	 * @throws Exception
	 */
	private static void startBackends(String[] args) throws Exception {
		/*
		 * 启动DM和SM的后台线程 启动心跳检测 启动拓扑同步服务 启动定制的后台服务
		 */
		if (AgentContext.getNodeRole() == ROLE.DM
				|| AgentContext.getNodeRole() == ROLE.SM) {
			HeartBeatWatcher.newInstance().startWatch();
			RMINetService.getInstance().bindService(new ClusterStatusMonitor(),
					ClusterConstants.CLUSTER_MONITOR_REF);
			RMINetService.getInstance().bindService(new LocalStatusMonitor(),
					ClusterConstants.LOCAL_MONITOR_REF);
		} else {
			RMILocalService.getInstance().bindService(new LocalStatusMonitor(),
					ClusterConstants.LOCAL_MONITOR_REF);
		}

		ClusterUtils.startCustomedBackends(args);

		// 下面启动关闭集群时需要清理或者关闭的线程——实现RunProcessInterface的关键服务
		String[] coreNames = new String[1];
		coreNames[0] = GeneralClusterManager.class.getSimpleName();

		// 真正的后台线程启动
		RunProcessInterface[] coreBackends = new RunProcessInterface[1];
		coreBackends[0] = new GeneralClusterManager();

		// 后台执行调度线程
		for (RunProcessInterface process : coreBackends) {
			process.callMain(args);
			RunProcessLauncher.addRunProcess2Cache(process);
		}

		// 监控是否所有的后台线程都起来了
		int initialized = 0;
		while (initialized < coreBackends.length) {
			for (int i = 0; i < coreBackends.length; i++) {
				if (coreBackends[i].isInitialized()) {
					initialized++;
					System.out.println("process [" + coreNames[i]
							+ "] started OK");
				}
			}
			ClusterUtils.doWait();
		}

	}

	/**
	 * 启动网络IO相关的线程:<br>
	 * 
	 * 1. 启动消息发送/接收/管理接口的服务
	 * 
	 * 2. 启动心跳检查和发送的巡逻器
	 * 
	 * 3. 如果是主节点，启动Web管理服务
	 * 
	 * @param args
	 * @throws Exception
	 */
	private static void startNetService(String[] args) throws Exception {

		List<RunProcessInterface> netServices = new ArrayList<RunProcessInterface>();

		List<String> netServiceNames = new ArrayList<String>();

		// 1. 启动消息发送/接收/管理接口的服务
		netServices.add(MessagingService.getInstance());
		netServiceNames.add(MessagingService.class.getSimpleName());

		// 2. 启动心跳检查和发送的巡逻器
		netServices.add(new HeartBeatGuard());
		netServiceNames.add(HeartBeatGuard.class.getSimpleName());

		for (RunProcessInterface process : netServices) {
			process.callMain(args);
			RunProcessLauncher.addRunProcess2Cache(process);
		}

		// 监控是否所有的后台线程都起来了
		int initialized = 0;
		while (initialized < netServices.size()) {
			for (int i = 0; i < netServices.size(); i++) {
				if (netServices.get(i).isInitialized()) {
					initialized++;
					System.out.println("process [" + netServiceNames.get(i)
							+ "] started OK");
				}
			}
			ClusterUtils.doWait();
		}

		// last. 如果是主节点，启动Web管理服务
		if (AgentContext.getNodeRole() == ROLE.DM) {
			//JettyServer.startJettyServer();
		}
	}

	/**
	 * 初始化上下文环境: 1. 日志配置 2. 集群参数导入
	 */
	private static void initContext(String[] args) throws JDOMException,
			IOException {
		configLogger();
		ClusterConfLoader.loadGlobalContext();
		for (String arg : args) {
			if (arg.equalsIgnoreCase("-jvm:debug")) {
				ClusterContext.setJvmDebugging(true);
				break;
			}
		}
	}

	/**
	 * 根据配置和命令行参数解析当前Agent在集群中角色
	 * 
	 * @param args
	 */
	private static void registerRoleInCluster(String[] args) {

		if (ClusterUtils.isLocalNode(ClusterContext.getDM())) {
			AgentContext.setNodeRole(ROLE.DM);
			String localIp = ClusterContext.getLocalAddress().getHostAddress();
			ClusterTopoDBService.getInstance().setDM(new Node(localIp));
			ClusterContext.setDM(localIp);
			return;
		}

		if (ClusterUtils.isLocalNode(ClusterContext.getBDM())) {
			AgentContext.setNodeRole(ROLE.BDM);
			String localIp = ClusterContext.getLocalAddress().getHostAddress();
			ClusterTopoDBService.getInstance().setBDM(new Node(localIp));
			ClusterContext.setBDM(localIp);
			return;
		}

		// 如果配置文件不能确定角色，从命令行获取信息
		for (String arg : args) {
			if (arg.equalsIgnoreCase("dm")) {
				AgentContext.setNodeRole(ROLE.DM);
				ClusterTopoDBService.getInstance().setDM(
						new Node(ClusterContext.getLocalAddress()
								.getHostAddress()));
				break;
			}
			if (arg.equalsIgnoreCase("bdm")) {
				AgentContext.setNodeRole(ROLE.BDM);
				ClusterTopoDBService.getInstance().setBDM(
						new Node(ClusterContext.getLocalAddress()
								.getHostAddress()));
				break;
			}
		}
	}

	private static void startMonitorModule(String[] args) {
		JobMonitor jobmonitor = new JobMonitor(true);
		jobmonitor.start();

		java.util.Timer timer = new java.util.Timer();
		timer.schedule(new UniversalMonitorMsgSender(), 10000, 10000);
	}

	/**
	 * 配置日志参数
	 */
	private static void configLogger() {
		org.apache.log4j.PropertyConfigurator.configure(ClusterContext
				.getRootDir() + logConfigFile);
	}
}

/**
 * 该Hook类处理: 系统关闭时，需要清理环境等
 * 
 */
class ShutdownHook extends Thread {
	private static Logger logger = LoggerFactory.getLogger(ShutdownHook.class);

	@Override
	public void run() {
		logger.info("received exit signal, begin to shutdown...");

		// 清理和关闭后台线程
		List<RunProcessInterface> processList = RunProcessLauncher
				.getRunProcessList();

		for (RunProcessInterface process : processList) {
			process.shutdown();
		}

		// 关闭启动的其他子进程Children
		JvmManager.closeAll();

		try {
			sleep(500L);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		logger.info("shutdown all backend processes OK");
		logger.info("Haflute Cluster Platform shutdown OK");
	}
}
